package com.song.shop.rocketmq;

import org.apache.rocketmq.client.consumer.AllocateMessageQueueStrategy;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.Set;

/**
 * @author Song Yinghui
 * @date 2018/8/24
 * @Version 1.0
 */
public class Customer2 {
    public static void main(String[] args) throws MQClientException {

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumerGroup");
        consumer.setNamesrvAddr("192.168.232.137:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        //consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("songtopic","*");


        consumer.setAllocateMessageQueueStrategy(new AllocateMessageQueueStrategy() {
            @Override
            public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll, List<String> cidAll) {
                return null;
            }

            @Override
            public String getName() {
                return null;
            }
        });
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                return null;
            }
        });
        consumer.registerMessageListener((MessageListenerConcurrently)(list , context)->{
            list.forEach(item ->{
                System.out.println("22222----"+ new SimpleDateFormat("yyyyMMdd HH:mm:ss").format(new Date()));
                System.out.println(new String(item.getBody()));
            });
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        /*consumer.registerMessageListener((MessageListenerOrderly) (list, consumeOrderlyContext) -> {
            consumeOrderlyContext.setAutoCommit(true);
            list.forEach( o -> {
                System.out.println("-----------");
                System.out.println(new String(o.getBody()));
            });
            return ConsumeOrderlyStatus.SUCCESS;
        });*/

        consumer.start();
    }

    public static void pullMessage() {
        DefaultMQPullConsumer pullConsumer = new DefaultMQPullConsumer();
        pullConsumer.setNamesrvAddr("111.231.232.212:9876");

        try {
            pullConsumer.start();
            Set<MessageQueue> messageQueues = pullConsumer.fetchSubscribeMessageQueues("songtopic");
            for (MessageQueue mq : messageQueues) {
                long l = pullConsumer.fetchConsumeOffset(mq, true);
                while (true){

                }
            }

        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}
